package com.ruoyi.swap.realtime.orderbook.controller;

import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.*;
import javax.servlet.http.HttpServletResponse;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.ruoyi.util.LarkDangerRobotUtils;
import com.ruoyi.util.LarkRecoverRobotUtils;
import com.ruoyi.util.WebSocketUtils;
import com.ruoyi.swap.constants.SwapMonitorConstants;
import com.ruoyi.swap.realtime.orderbook.domain.VO.SwapOrderbookRealtimeDataVO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import com.ruoyi.common.annotation.Log;
import com.ruoyi.common.core.controller.BaseController;
import com.ruoyi.common.core.domain.AjaxResult;
import com.ruoyi.common.enums.BusinessType;
import com.ruoyi.swap.realtime.orderbook.domain.SwapOrderbookRealtimeData;
import com.ruoyi.swap.realtime.orderbook.service.ISwapOrderbookRealtimeDataService;
import com.ruoyi.common.utils.poi.ExcelUtil;
import com.ruoyi.common.core.page.TableDataInfo;

/**
 * 实时订单簿Controller
 *
 * @author saumell
 * @date 2024-08-03
 */
@RestController
@RequestMapping("/swaprealtimeorderbook/swaprealtimeorderbookbak")
public class SwapOrderbookRealtimeDataController extends BaseController {
    
    private static final Logger log = LoggerFactory.getLogger(SwapOrderbookRealtimeDataController.class);
    
    @Autowired
    private ISwapOrderbookRealtimeDataService swapOrderbookRealtimeDataService;

    @Value("${my.lark.dangerlark}")
    private String dangerlark;

    @Value("${my.lark.recoverlark}")
    private String recoverlark;

    /**
     * 查询实时订单簿列表
     */
    @PreAuthorize("@ss.hasPermi('swaprealtimeorderbook:swaprealtimeorderbookbak:list')")
    @GetMapping("/list")
    public TableDataInfo list(SwapOrderbookRealtimeDataVO swapOrderbookRealtimeData) {
        startPage();
        List<SwapOrderbookRealtimeDataVO> list = swapOrderbookRealtimeDataService.selectSwapOrderbookRealtimeDataList(swapOrderbookRealtimeData);
        return getDataTable(list);
    }

    /**
     * 导出实时订单簿列表
     */
    @PreAuthorize("@ss.hasPermi('swaprealtimeorderbook:swaprealtimeorderbookbak:export')")
    @Log(title = "实时订单簿", businessType = BusinessType.EXPORT)
    @PostMapping("/export")
    public void export(HttpServletResponse response, SwapOrderbookRealtimeDataVO swapOrderbookRealtimeData) {
        List<SwapOrderbookRealtimeDataVO> list = swapOrderbookRealtimeDataService.selectSwapOrderbookRealtimeDataList(swapOrderbookRealtimeData);
        ExcelUtil<SwapOrderbookRealtimeDataVO> util = new ExcelUtil<SwapOrderbookRealtimeDataVO>(SwapOrderbookRealtimeDataVO.class);
        util.exportExcel(response, list, "实时订单簿数据");
    }

    /**
     * 获取实时订单簿详细信息
     */
    @PreAuthorize("@ss.hasPermi('swaprealtimeorderbook:swaprealtimeorderbookbak:query')")
    @GetMapping(value = "/{id}")
    public AjaxResult getInfo(@PathVariable("id") Integer id) {
        return success(swapOrderbookRealtimeDataService.selectSwapOrderbookRealtimeDataById(id));
    }

    /**
     * 接收返回值
     *
     * @param wsClient
     * @param subParam
     * @return
     * @throws InterruptedException
     */
    private String handleSubscription(WebSocketUtils wsClient, String subParam) throws InterruptedException {
        wsClient.send(subParam);
        long startTime = System.currentTimeMillis();
        long endTime = startTime + 6000; // 6秒超时
        String response = null;
        int messageCount = 0; // 添加计数器

        while (System.currentTimeMillis() < endTime) {
            Thread.sleep(200);
            String message = wsClient.getMsg();

            if (message != null && !message.isEmpty()) {
                messageCount++; // 每次收到消息时增加计数

                if (message.contains("ping")) {
                    wsClient.send(message.replace("ping", "pong"));
                } else {
                    if (messageCount > 1) { // 当计数器大于1时处理消息
                        response = message;
                        break;
                    }
                }
                // 清除处理过的消息
                wsClient.clearMsg();
            }
        }

        return response;
    }


    /**
     * 判断返回值状态
     *
     * @param websocketResponse
     * @return
     */
    private static String determineReturnValue(String websocketResponse) {
        try {
            JSONObject jsonObject = JSON.parseObject(websocketResponse);
            JSONObject tick = jsonObject.getJSONObject("tick");
            JSONArray asks = tick.getJSONArray("asks");
            JSONArray bids = tick.getJSONArray("buys");
            // 检查asks是否为空或者内容是否为[0,0]
            if (asks == null || asks.isEmpty() || (asks.size() == 1 && asks.getJSONArray(0).equals(Arrays.asList(0, 0)))) {
                if (bids == null || bids.isEmpty() || (bids.size() == 1 && bids.getJSONArray(0).equals(Arrays.asList(0, 0)))) {
                    return "false, Asks和Bids均未获取到数据";
                }
                return "false, Asks未获取到数据";
            }
            // 检查bids是否为空或者内容是否为[0,0]
            if (bids == null || bids.isEmpty() || (bids.size() == 1 && bids.getJSONArray(0).equals(Arrays.asList(0, 0)))) {
                return "false, Bids未获取到数据";
            }
            return "true";
        } catch (Exception e) {
            return "false, JSON解析错误！";
        }
    }

    /**
     * 通过Websocket实时订阅，监控合约Orderbook返回值内容
     */

    public void swapOrderBookMonitor() {
        List<SwapOrderbookRealtimeDataVO> swapOrderbookCurrencysList = swapOrderbookRealtimeDataService.selectSwapAllData();
        //新建放置需要报警、故障修复交易对的id
        List<Integer> dangerProductionInspections = new ArrayList<Integer>();
        List<Integer> recoverProductionInspections = new ArrayList<Integer>();
        for (SwapOrderbookRealtimeDataVO swapOrderbookRealtimeDataVO : swapOrderbookCurrencysList) {
            WebSocketUtils wsClient = null;
            Integer id = swapOrderbookRealtimeDataVO.getId();
            String currencyPair = swapOrderbookRealtimeDataVO.getCurrencyPair();
            String callingState = swapOrderbookRealtimeDataVO.getCallingState();
            Date lastExecutionTime = swapOrderbookRealtimeDataVO.getLastExecutionTime();
            Integer successTimes = swapOrderbookRealtimeDataVO.getSuccessTimes();
            Integer failTimes = swapOrderbookRealtimeDataVO.getFailTimes();
            Integer dangerWarning = swapOrderbookRealtimeDataVO.getDangerWarning();
            Integer safeWarning = swapOrderbookRealtimeDataVO.getSafeWarning();
            String websocketResponse = null;
            try {
                // 在每次循环中建立 WebSocket 连接
                wsClient = WebSocketUtils.executeWebSocket(SwapMonitorConstants.SWAP_WEBSOCKET_ORDERBOOK_URL);
                String subParam = "{\"event\":\"sub\",\"params\":{\"channel\":\"market_" + currencyPair + "_depth_step0\",\"cb_id\":\"1\"}}";
                // 发送订阅消息
                websocketResponse = handleSubscription(wsClient, subParam);
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                // 确保在每次循环结束后关闭 WebSocket 连接
                if (wsClient != null && wsClient.isOpen()) {
                    wsClient.close();
                }
            }
            // 处理返回的响应
            if (websocketResponse == null || websocketResponse.isEmpty()) {
                //                合约Orderbook-未接收到数据
                if (Objects.isNull(callingState) || callingState.isEmpty()) {
                    //数据库中没有数据，说明是第一次订阅即失败
                    //将调用失败的id放入dangerProductionInspections中
                    LocalDateTime currentDateTime = LocalDateTime.now();
                    lastExecutionTime = Timestamp.valueOf(currentDateTime);
                    successTimes = 0;
                    failTimes = 1;
                    Integer currencysId = swapOrderbookRealtimeDataService.queryCurryNumber(currencyPair);
                    SwapOrderbookRealtimeData o1 = new SwapOrderbookRealtimeData(id, currencysId, "off_line", lastExecutionTime, successTimes, failTimes, "Server未推送Orderbook数据", null, null, null);
                    swapOrderbookRealtimeDataService.insertSwapOrderbookRealtimeData(o1);
                    dangerProductionInspections.add(o1.getId());
                    log.info("[合约Orderbook-1}：【" + currencyPair + "】首次调用即失败");
                } else if (callingState.equals("off_line")) {
                    //订阅一直失败
                    //将调用失败的id放入dangerProductionInspections中
                    if (dangerWarning == null || dangerWarning == 1) {
                        dangerProductionInspections.add(id);
                    }
                    LocalDateTime currentDateTime = LocalDateTime.now();
                    lastExecutionTime = Timestamp.valueOf(currentDateTime);
                    failTimes = failTimes + 1;
                    SwapOrderbookRealtimeData o1 = new SwapOrderbookRealtimeData(id, null, callingState, lastExecutionTime, successTimes, failTimes, "Server未推送Orderbook数据", null, null, null);
                    swapOrderbookRealtimeDataService.updateSwapOrderbookRealtimeData(o1);
                    log.info("[合约Orderbook-2}：【" + currencyPair + "】订阅一直失败");
                } else if (callingState.equals("on_line")) {
                    //此种情况为第一次订阅失败
                    //将调用失败的id放入dangerProductionInspections中
                    dangerProductionInspections.add(id);
                    callingState = "off_line";
                    LocalDateTime currentDateTime = LocalDateTime.now();
                    lastExecutionTime = Timestamp.valueOf(currentDateTime);
                    failTimes = failTimes + 1;
                    SwapOrderbookRealtimeData o1 = new SwapOrderbookRealtimeData(id, null, callingState, lastExecutionTime, successTimes, failTimes, "Server未推送Orderbook数据", null, null, null);
                    swapOrderbookRealtimeDataService.updateSwapOrderbookRealtimeData(o1);
                    log.info("[合约Orderbook-3}：【" + currencyPair + "】订阅首次失败");
                } else {
                    log.info("+++++++++++++++++++++++++++订阅失败未知情况+++++++++++++++++++++++++++++++");
                    log.info("[合约Orderbook-4}：【" + currencyPair + "】未知情况");
                }
            }else if (websocketResponse != null && determineReturnValue(websocketResponse) != null && determineReturnValue(websocketResponse).startsWith("false")) {
                // 处理返回值包含 "false" 的情况
                if (Objects.isNull(callingState) || callingState.isEmpty()) {
                    //数据库中没有数据，说明是第一次订阅即失败
                    //将调用失败的id放入dangerProductionInspections中
                    LocalDateTime currentDateTime = LocalDateTime.now();
                    lastExecutionTime = Timestamp.valueOf(currentDateTime);
                    successTimes = 0;
                    failTimes = 1;
                    Integer currencysId = swapOrderbookRealtimeDataService.queryCurryNumber(currencyPair);
                    SwapOrderbookRealtimeData o1 = new SwapOrderbookRealtimeData(id, currencysId, "off_line", lastExecutionTime, successTimes, failTimes, determineReturnValue(websocketResponse).split(", ")[1], null, null, null);
                    swapOrderbookRealtimeDataService.insertSwapOrderbookRealtimeData(o1);
                    dangerProductionInspections.add(o1.getId());
                    log.info("[合约Orderbook-5}：【" + currencyPair + "】" + determineReturnValue(websocketResponse));
                } else if (callingState.equals("off_line")) {
                    //订阅一直失败
                    //将调用失败的id放入dangerProductionInspections中
                    if (dangerWarning == null || dangerWarning == 1) {
                        dangerProductionInspections.add(id);
                    }
                    LocalDateTime currentDateTime = LocalDateTime.now();
                    lastExecutionTime = Timestamp.valueOf(currentDateTime);
                    failTimes = failTimes + 1;
                    SwapOrderbookRealtimeData o1 = new SwapOrderbookRealtimeData(id, null, callingState, lastExecutionTime, successTimes, failTimes, determineReturnValue(websocketResponse).split(", ")[1], null, null, null);
                    swapOrderbookRealtimeDataService.updateSwapOrderbookRealtimeData(o1);
                    log.info("[合约Orderbook-6}：【" + currencyPair + "】" + determineReturnValue(websocketResponse));
                } else if (callingState.equals("on_line")) {
                    //此种情况为第一次订阅失败
                    //将调用失败的id放入dangerProductionInspections中
                    dangerProductionInspections.add(id);
                    callingState = "off_line";
                    LocalDateTime currentDateTime = LocalDateTime.now();
                    lastExecutionTime = Timestamp.valueOf(currentDateTime);
                    failTimes = failTimes + 1;
                    SwapOrderbookRealtimeData o1 = new SwapOrderbookRealtimeData(id, null, callingState, lastExecutionTime, successTimes, failTimes, determineReturnValue(websocketResponse).split(", ")[1], null, null, null);
                    swapOrderbookRealtimeDataService.updateSwapOrderbookRealtimeData(o1);
                    log.info("[合约Orderbook-7}：【" + currencyPair + "】" + determineReturnValue(websocketResponse));
                } else {
                    log.info("+++++++++++++++++++++++++++订阅失败未知情况+++++++++++++++++++++++++++++++");
                    log.info("[合约Orderbook-8}：【" + currencyPair + "】" + determineReturnValue(websocketResponse));
                }
            }else if (websocketResponse != null && determineReturnValue(websocketResponse) != null && determineReturnValue(websocketResponse).equals("true")) {
//             订阅+格式校验均成功！
                if (Objects.isNull(callingState) || callingState.isEmpty()) {
//                    首次订阅
                    LocalDateTime currentDateTime = LocalDateTime.now();
                    lastExecutionTime = Timestamp.valueOf(currentDateTime);
                    successTimes = 1;
                    failTimes = 0;
                    Integer currencysId = swapOrderbookRealtimeDataService.queryCurryNumber(currencyPair);
                    SwapOrderbookRealtimeData o1 = new SwapOrderbookRealtimeData(id, currencysId, "on_line", lastExecutionTime, successTimes, failTimes, null, null, null, null);
                    swapOrderbookRealtimeDataService.insertSwapOrderbookRealtimeData(o1);
                    log.info("[合约Orderbook-9}：【" + currencyPair + "】" + determineReturnValue(websocketResponse));
                } else if (callingState.equals("on_line")) {
//                    订阅一直成功
                    LocalDateTime currentDateTime = LocalDateTime.now();
                    lastExecutionTime = Timestamp.valueOf(currentDateTime);
                    successTimes = successTimes + 1;
                    SwapOrderbookRealtimeData o1 = new SwapOrderbookRealtimeData(id, null, callingState, lastExecutionTime, successTimes, failTimes, null, null, null, null);
                    swapOrderbookRealtimeDataService.updateSwapOrderbookRealtimeData(o1);
                    log.info("[合约Orderbook-10}：【" + currencyPair + "】" + determineReturnValue(websocketResponse));
                } else if (callingState.equals("off_line")) {
//                    订阅故障修复，复通
                    if (safeWarning == null || safeWarning == 1) {
                        recoverProductionInspections.add(id);
                    }
                    callingState = "on_line";
                    LocalDateTime currentDateTime = LocalDateTime.now();
                    lastExecutionTime = Timestamp.valueOf(currentDateTime);
                    successTimes = successTimes + 1;
                    SwapOrderbookRealtimeData o1 = new SwapOrderbookRealtimeData(id, null, callingState, lastExecutionTime, successTimes, failTimes, null, null, null, null);
                    swapOrderbookRealtimeDataService.updateSwapOrderbookRealtimeData(o1);
                    log.info("[合约Orderbook-11}：【" + currencyPair + "】" + determineReturnValue(websocketResponse));
                } else {
                    log.info("+++++++++++++++++++++++++++订阅成功未知情况+++++++++++++++++++++++++++++++");
                    log.info("[合约Orderbook-12}：【" + currencyPair + "】" + determineReturnValue(websocketResponse));
                }
            }
        }
        List<SwapOrderbookRealtimeDataVO> swapOrderbookCurrencysList_new = swapOrderbookRealtimeDataService.selectSwapAllData();
        //        dangerProductionInspections :断开交易对对应id
        //        recoverProductionInspections  ：复通交易对对应id
        //        需要发送危险报警对象
        List<SwapOrderbookRealtimeDataVO> allProductionInspections2 = new ArrayList<>();
        //        需要发送复通信息对象
        List<SwapOrderbookRealtimeDataVO> allProductionInspections3 = new ArrayList<>();
        for (int i = 0; i < swapOrderbookCurrencysList_new.size(); i++) {
            Integer id = swapOrderbookCurrencysList_new.get(i).getId();
            if (dangerProductionInspections.contains(id)) {
                allProductionInspections2.add(swapOrderbookCurrencysList_new.get(i));
            } else if (recoverProductionInspections.contains(id)) {
                allProductionInspections3.add(swapOrderbookCurrencysList_new.get(i));
            }
        }
        if (!allProductionInspections2.isEmpty()) {
            if (allProductionInspections2.size() == 1) {
                // 如果只有一条数据
                String name = allProductionInspections2.get(0).getCurrencyPair();
                Date time = allProductionInspections2.get(0).getLastExecutionTime();
                String reason = allProductionInspections2.get(0).getLastFailureReason();
                String formattedDate = "";
                if (time != null) {
                    SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    formattedDate = formatter.format(time);
                }
                // 拼接调用失败报警信息...
                String DangerLarkNotice = "【合约-Orderbook-WebSocket】下方为Orderbook故障的交易对名称及调用时间，请您关注！\n";
                DangerLarkNotice += "交易对名称： " + name + "   " + "故障时间： " + formattedDate + "    故障原因： " + reason + "   " + "\n";
                String dangernotice = LarkDangerRobotUtils.notice(DangerLarkNotice, dangerlark);
                // 更新数据库状态
                Integer id = allProductionInspections2.get(0).getId();
                String statusMessage = dangernotice.equals("lark_success") ? "Lark发送故障X信息情况成功！" : "Lark发送故障X信息情况失败！";
                SwapOrderbookRealtimeData o2 = new SwapOrderbookRealtimeData(id, null, null, null, null, null, null, dangernotice.equals("lark_success") ? 0 : 1, null, statusMessage);
                swapOrderbookRealtimeDataService.updateSwapOrderbookRealtimeData(o2);
//                }
            } else {
                // 如果有多条数据，正常调用notice
                String DangerLarkNotice = "【合约-Orderbook-WebSocket】下方为Orderbook故障的交易对名称及调用时间，请您关注！\n";
                boolean needNotice = false; // 是否需要发送通知
                for (int i = 0; i < allProductionInspections2.size(); i++) {
                    String name = allProductionInspections2.get(i).getCurrencyPair();
                    Date time = allProductionInspections2.get(i).getLastExecutionTime();
                    String reason = allProductionInspections2.get(i).getLastFailureReason();
                    String formattedDate = "";
                    if (time != null) {
                        SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        formattedDate = formatter.format(time);
                    }
                    DangerLarkNotice += "交易对名称： " + name + "   " + "故障时间： " + formattedDate + "    故障原因： " + reason + "   " + "\n";
                }
//                if (needNotice) {
                String dangernotice = LarkDangerRobotUtils.notice(DangerLarkNotice, dangerlark);
                // 检查通知是否成功发送
                if (dangernotice.equals("lark_success")) {
                    // 更新数据库状态
                    for (int i = 0; i < allProductionInspections2.size(); i++) {
                        Integer id = allProductionInspections2.get(i).getId();
                        String statusMessage = "Lark发送故障X信息情况成功！";
                        SwapOrderbookRealtimeData o2 = new SwapOrderbookRealtimeData(id, null, null, null, null, null, null, 0, null, statusMessage);
                        swapOrderbookRealtimeDataService.updateSwapOrderbookRealtimeData(o2);
                    }
                } else {
                    // 更新数据库状态
                    for (int i = 0; i < allProductionInspections2.size(); i++) {
                        Integer id = allProductionInspections2.get(i).getId();
                        String statusMessage = "Lark发送故障X信息情况失败！";
                        SwapOrderbookRealtimeData o2 = new SwapOrderbookRealtimeData(id, null, null, null, null, null, null, 1, null, statusMessage);
                        swapOrderbookRealtimeDataService.updateSwapOrderbookRealtimeData(o2);
                    }
                }
            }
        }
        if (allProductionInspections3.size() != 0) {
//            拼接重新调用成功对象
            String RecoverLarkNotice = "";
            RecoverLarkNotice += "【合约-Orderbook-WebSocket】下方为K恢复的交易对名称及调用时间，请您关注！" + "\n";
            for (int i = 0; i < allProductionInspections3.size(); i++) {
                String name = allProductionInspections3.get(i).getCurrencyPair();
                Date time = allProductionInspections3.get(i).getLastExecutionTime();
                String formattedDate = "";
                if (time != null) {
                    SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    formattedDate = formatter.format(time);
                }
                RecoverLarkNotice += "交易对名称： " + name + "   " + "恢复时间： " + formattedDate + "   " + "\n";
            }
            String recovernotice = LarkRecoverRobotUtils.notice(RecoverLarkNotice, recoverlark);
            for (int i = 0; i < allProductionInspections3.size(); i++) {
                //判断复通信息通知情况
                if (recovernotice.equals("lark_fail")) {
//                    通知失败
                    for (int j = 0; j < allProductionInspections3.size(); j++) {
                        Integer id = allProductionInspections3.get(j).getId();
                        SwapOrderbookRealtimeData o2 = new SwapOrderbookRealtimeData(id, null, null, null, null, null, null, null, 1, "Lark发送故障V信息情况失败！");
                        swapOrderbookRealtimeDataService.updateSwapOrderbookRealtimeData(o2);
                    }
                } else {
//                    通知成功
                    for (int k = 0; k < allProductionInspections3.size(); k++) {
                        Integer id = allProductionInspections3.get(k).getId();
                        SwapOrderbookRealtimeData o2 = new SwapOrderbookRealtimeData(id, null, null, null, null, null, null, null, 0, "Lark发送故障V信息情况失败！");
                        swapOrderbookRealtimeDataService.updateSwapOrderbookRealtimeData(o2);
                    }
                }
            }
        }
        log.info("============【【合约-Orderbook-WebSocket】执行完成============");


































    }
}
